package com.blog.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.blog.pojo.CacheProperties;
import com.google.protobuf.InvalidProtocolBufferException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class CanalListener {
    @Autowired
    private RedisTemplate redisTemplate;
    public void run(){
        System.out.println("listener运行");
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("localhost", 11111),
                "example", "", ""
        );
        int batchSize=1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                try {
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        Thread.sleep(1000);
                    } else {
                        dataHandle(message.getEntries());
                    }
                    connector.ack(batchId);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            connector.disconnect();
        }
    }

    private void dataHandle(List<CanalEntry.Entry> entries) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entries) {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType();
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            String tableName = entry.getHeader().getTableName();
            if (eventType.equals(CanalEntry.EventType.DELETE)) {
                    delete(rowDatasList,tableName);
            }
            if (eventType.equals(CanalEntry.EventType.UPDATE)){
                update(rowDatasList,tableName);
            }
        }
    }

    private void update(List<CanalEntry.RowData> rowDatasList, String tableName) {
        if (tableName.equals("article")) {
            for (CanalEntry.RowData rowData : rowDatasList) {
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                HashMap<String, Object> map = new HashMap<>();
                for (CanalEntry.Column column : afterColumnsList) {
                    map.put(column.getName(),column.getValue());
                }
                String jsonString = JSON.toJSONString(map);
                String id = (String) map.get("id");
                redisTemplate.opsForValue().set(CacheProperties.PRE_ARTICLE_KEY+id,jsonString,1, TimeUnit.HOURS);
            }
        }
    }

    private void delete(List<CanalEntry.RowData> rowDatasList, String tableName) {
        if (tableName.equals("article")) {
            for (CanalEntry.RowData rowData : rowDatasList) {
                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                for (CanalEntry.Column column : beforeColumnsList) {
                    if (column.getName().equals("id")) {
                        redisTemplate.delete(CacheProperties.PRE_ARTICLE_KEY + column.getValue());
                    }
                }
            }
        }
    }

}
